import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.trees.TreePattern.AND
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.types.{DataTypes, IntegerType, StringType, StructField, StructType}

object data1_SQL3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("Spark Pi")
      .getOrCreate()
    val sc =spark.sparkContext
    //读取电影用户信息
    val schemaUser = StructType(Seq(
      StructField("id",IntegerType),
      StructField("gender",StringType),
      StructField("age", IntegerType),
      StructField("zipcode", StringType)
    ))
    val user=spark.read.option("sep","::").schema(schemaUser)
      .csv("F:\\Spark\\karry\\karry\\src\\main\\resources\\users.dat")
    user.show(5)
    // 1查询
    user.where("gender='F' and age = 18").show(5)
    user.filter("gender ='F' and age = 18").show(5)
    spark.udf.register("replace",(x:String) =>{
      x match {
        case "M" =>0
        case "F" =>1
      }
    })
    user.selectExpr("id","replace(gender) as sexual","age").show(3)
    user.select(user.col("zipcode")).collect().foreach(println)
    user.select(user.apply("zipcode")).collect().foreach(println)

    //排序
 // user.orderBy(desc("id")).show(5)
    user.orderBy(-user("id")).show(5)
    user.sort(desc("id")).show(5)
//分组
    user.groupBy("gender").count().show()
    val schemaRatings = StructType(Seq(
      StructField("user_id", IntegerType),
      StructField("movie_id", IntegerType),
      StructField("rating", IntegerType),
      StructField("timestamp", StringType)
    ))
    val ratings = spark.read
      .option("sep", "::")
      .schema(schemaRatings)
      .csv("F:\\Spark\\karry\\karry\\src\\main\\resources\\ratings.dat")

    val result = user
      .join(ratings, user("id") === ratings("user_id"))
      .where(user("age") === 18 && user("gender") === "F" && ratings("rating") === 5)
      .select("movie_id")
      .distinct()

    println("18岁女生评分为5分的电影ID：")
    result.show()


    sc.stop()

  }


}
